package my.thesis.spoutsReliable;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

import java.net.URL;

import java.util.HashMap;
import java.util.Map;

import org.joda.time.DateTime;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

public class TweetReaderUrlReliable extends BaseRichSpout
{
	private static final long serialVersionUID = 5773846193661629157L;
	private BufferedReader reader;
	private SpoutOutputCollector _collector;
	
	private final String URL = "http://minux.web4nuts.com:55555/decode_2012-09-24_06-15.txt?1";
	
	private Map<Integer, String> SpoutPendingTuples; //#non-acked tuples = #spout pending tuples
	private Map<Integer, String> toSend;
	
	private Map<Integer, Long> SpoutPendingTuplesTimestamps; //#non-acked tuples = #spout pending tuples
	private Map<Integer, Long> toSendTimestamps;
	
	private int failedRetries=0;

	int i=1; //using for msgID

	public void nextTuple() 
	{
		try {
			String in;
			in = this.reader.readLine();
			if(in!=null)
			{
				DateTime timestamp = DateTime.now();
				long millis = timestamp.getMillis();
				
					SpoutPendingTuples.put(i, in);
					toSend.put(i, in);
					
					SpoutPendingTuplesTimestamps.put(i, millis);
					toSendTimestamps.put(i, millis);
					
			}
			else 
			{
					//System.out.println("Nothing to read, wait 10 sec.");
					//Thread.sleep(10000);
					
					failedRetries++;
					if(failedRetries > 10)
					{
						System.out.println("End of file reached bb...");
					}
					
					//sleep gia 5 ms
					Thread.sleep(10);	
			}
			
			for(Map.Entry<Integer, String> entry : toSend.entrySet())
			{
				_collector.emit(new Values(entry.getValue(),toSendTimestamps.get(entry.getKey())),entry.getKey()); //emit all the tuples from the toSend HashMap
				
			}
			toSend.clear();				
			toSendTimestamps.clear();
			//clear the HashMap
		} 
		catch (Exception e)
		{
			System.err.println("Error reading line "+e.getMessage());
		}
		
		i += 1;
	}

	public void open(Map conf, TopologyContext context,SpoutOutputCollector collector) 
	{	
		this._collector = collector;
		
		this.SpoutPendingTuples = new HashMap<Integer, String>();
		this.toSend = new HashMap<Integer, String>();
		
		this.SpoutPendingTuplesTimestamps = new HashMap<Integer,Long>();
		this.toSendTimestamps=new HashMap<Integer,Long>();
		
		try 
		{
			URL url = new URL(URL);
			this.reader = new BufferedReader(new InputStreamReader(url.openStream()));	
		}
		catch (IOException e)
		{
			System.err.println("Error in communication with server ["+e.getMessage()+"]");
		}
	}
	
    public void ack(Object id) 
	{
		System.out.println("OK ack for "+id+" Size : "+SpoutPendingTuples.size());
		SpoutPendingTuples.remove(id);
		SpoutPendingTuplesTimestamps.remove(id);
		//Remove the tuple acked from the Map	
    }

    public void fail(Object id) 
    {
    	Integer msgID = (Integer) id;
    	toSend.put(msgID, SpoutPendingTuples.get(msgID));
    	toSendTimestamps.put(msgID,SpoutPendingTuplesTimestamps.get(msgID));
    	//Resend the tuple that failed
    }

	public void declareOutputFields(OutputFieldsDeclarer declarer) 
	{
		declarer.declare(new Fields("jsonString","timestamp"));
	}
}
